package queue

import (
	"context"
	"sync"
	"time"
)

type Delayable interface {
	Delay() time.Duration
}

// 结合优先级队列、并发阻塞队列实现延时队列
type DelayQueue[T Delayable] struct {
	dequeSignal chan struct{}
	enqueSignal chan struct{}

	mu sync.Mutex
	// 用优先级队列做底层存储
	pq   *PriorityQueue[T]
	zero T
}

func NewDelayQueue[T Delayable](capacity int) *DelayQueue[T] {
	return &DelayQueue[T]{
		// 把 Delayable 和 Comparator 结合起来了
		pq: NewPriorityQueue[T](capacity, func(x, y T) int {
			if x.Delay() > y.Delay() {
				return 1
			} else if x.Delay() == y.Delay() {
				return 0
			} else {
				return -1
			}
		}),
		dequeSignal: make(chan struct{}),
		enqueSignal: make(chan struct{}),
	}
}

// BenchmarkDelayQueue-20    	     577	   1836298 ns/op	   27880 B/op	     575 allocs/op
func (c *DelayQueue[T]) EnQueue(ctx context.Context, data T) error {
	c.mu.Lock()
	for c.pq.IsFull() {
		c.mu.Unlock()
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-c.dequeSignal:
			c.mu.Lock()
		}
	}
	_ = c.pq.EnQueue(data)
	c.mu.Unlock()
	select {
	case <-c.enqueSignal:
	default:
	}
	return nil
}
func (c *DelayQueue[T]) DeQueue(ctx context.Context) (T, error) {
	var timer *time.Timer
	for {
		c.mu.Lock()
		// 队列空处理
		for c.pq.IsEmpty() {
			c.mu.Unlock()
			select {
			case <-ctx.Done():
				return c.zero, ctx.Err()
			case <-c.enqueSignal:
				c.mu.Lock()
			}
		}

		val, _ := c.pq.Peek()
		// 队首元素已到期处理
		if val.Delay() <= 0 {
			val, _ = c.pq.DeQueue()
			c.mu.Unlock()
			select {
			case <-c.dequeSignal:
			default:
			}
			return val, nil
		}

		// 队首元素未到期处理
		if timer == nil {
			timer = time.NewTimer(val.Delay())
		} else {
			timer.Reset(val.Delay())
		}
		c.mu.Unlock()
		select {
		case <-ctx.Done():
			return c.zero, ctx.Err()
		case <-c.enqueSignal:
		case <-timer.C:
		}
	}
}

// BenchmarkDelayQueue-20    	     619	   1782604 ns/op	   36700 B/op	     667 allocs/op
func (c *DelayQueue[T]) EnQueueV1(ctx context.Context, data T) error {
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		default:
		}

		c.mu.Lock()

		err := c.pq.EnQueue(data)
		switch err {
		case nil:
			// 入队成功
			// 队列从空到有一个才发信号
			if c.pq.Len() == 1 {
				close(c.enqueSignal)
				c.enqueSignal = make(chan struct{})
			}

			c.mu.Unlock()
			return nil
		case ErrOutOfCapacity:
			// 队满
			c.mu.Unlock()
			select {
			case <-ctx.Done():
				return ctx.Err()
			case <-c.dequeSignal:
			}
		default:
			c.mu.Unlock()
			return err
		}
	}
}
func (c *DelayQueue[T]) DeQueueV1(ctx context.Context) (T, error) {
	var timer *time.Timer
	for {
		select {
		case <-ctx.Done():
			return c.zero, ctx.Err()
		default:
		}

		c.mu.Lock()

		val, err := c.pq.Peek()
		switch err {
		case nil:
			delay := val.Delay()
			if delay <= 0 {
				isFull := c.pq.IsFull()
				val, _ := c.pq.DeQueue()
				// 队列从满到空一个才发信号
				if isFull {
					close(c.dequeSignal)
					c.dequeSignal = make(chan struct{})
				}
				c.mu.Unlock()
				return val, nil
			}

			if timer == nil {
				timer = time.NewTimer(delay)
			} else {
				timer.Reset(delay)
			}
			c.mu.Unlock()
			select {
			case <-timer.C:
			case <-c.enqueSignal: // 同时也要监听从满到空一个信号
			case <-ctx.Done():
				return c.zero, ctx.Err()
			}
		case ErrEmptyQ:
			c.mu.Unlock()
			select {
			case <-c.enqueSignal:
			case <-ctx.Done():
				return c.zero, ctx.Err()
			}
		default:
			c.mu.Unlock()
			return c.zero, err
		}
	}
}
func (c *DelayQueue[T]) IsFull() bool {
	return c.pq.IsFull()
}
func (c *DelayQueue[T]) IsEmpty() bool {
	return c.pq.IsEmpty()
}
func (c *DelayQueue[T]) Len() uint64 {
	return uint64(c.pq.Len())
}
